package org.pupil.register.zookeeper;

import com.alibaba.fastjson.JSON;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.pupil.common.Cache;
import org.pupil.common.URL;
import org.pupil.event.*;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

import static org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent.Type.*;

/**
 * @Description: Zookeeper注册中心实现层，具体的功能逻辑实现
 * @Author: pupil
 * @Date: 2024/07/10 下午 10:09
 */
public class CuratorZookeeperRegistry extends AbstractZookeeperRegistry {

    // 连接失败等待重试时间
    public static final int BASE_SLEEP_TIME_MS = 1000;
    // 重试次数
    public static final int MAX_RETRIES = 3;
    // 根路径
    public static final String ROOT_PATH = "/pupil_rpc";
    // 提供者路径
    public static final String PROVIDER = "/provider";

    public static final ConcurrentHashMap<String,  PathChildrenCache> PATH_CHILDREN_CACHE_MAP = new ConcurrentHashMap<>();

    // 与 ZooKeeper 集群交互的客户端
    private final CuratorFramework client;

    /**
     * 启动zk
     *
     * @param registerAddr 注册中心ip地址和端口
     */
    public CuratorZookeeperRegistry(String registerAddr) {
        // 通过工厂创建CuratorFramework的实例，     ZooKeeper服务器的地址        重试策略
        client = CuratorFrameworkFactory.newClient(registerAddr, new ExponentialBackoffRetry(BASE_SLEEP_TIME_MS, MAX_RETRIES));
        // 启动zk
        client.start();
    }

    /**
     * 服务注册
     * ROOT_PATH : 根路径
     * ROOT_PATH + Service + $ version : 监听
     * ROOT_PATH + Service + $ version + Host + Port : 节点
     *
     * @param url 服务的资源定位符
     * @throws Exception
     */
    public void register(URL url) throws Exception {
        // 判断根节点是否存在
        if (!existNode(ROOT_PATH)) {
            // 创建持久化根节点
            client.create().creatingParentContainersIfNeeded()
                    .withMode(CreateMode.PERSISTENT).forPath(ROOT_PATH, "".getBytes());
        }

        final String providerDataPath = getProviderDataPath(url);

        // 判断服务节点是否存在，存在则删除节点
        if (existNode(providerDataPath)) {
            // 删除节点
            deleteNode(providerDataPath);
        }
        // 创建节点
        client.create().creatingParentContainersIfNeeded()
                .withMode(CreateMode.EPHEMERAL).forPath(providerDataPath, JSON.toJSONString(url).getBytes());
        // 更新本地缓存
        super.register(url);
    }

    /**
     * 取消服务注册（服务销毁）
     *
     * @param url 服务的资源定位符
     * @throws Exception 异常
     */
    public void unRegister(URL url) throws Exception {
        // 删除服务节点
        deleteNode(getProviderDataPath(url));
        // 更新本地缓存
        super.unRegister(url);
    }

    /**
     * 服务发现
     *
     * @param serviceName 服务名
     * @param version     版本号
     * @return 服务提供方集合
     * @throws Exception 异常
     */
    public List<URL> discoveries(String serviceName, String version) throws Exception {
        // 从本地缓存获取服务集合
        List<URL> urls = super.discoveries(serviceName, version);
        // 判断集合是否为空
        if (null == urls || urls.isEmpty()) {
            // 获取该服务下的所有服务提供方集合
            final List<String> strings = client.getChildren().forPath(getProviderPath(serviceName, version));
            // 判断集合是否为空
            if (!strings.isEmpty()) {
                // 读取服务节点信息创建新的集合
                urls = new ArrayList<>();
                for (String string : strings) {
                    final String[] split = string.split(":");
                    urls.add(new URL(split[0], Integer.parseInt(split[1])));
                }
            }
        }
        return urls;
    }

    /**
     * 服务订阅
     *
     * @param url 服务提供方资源定位符
     * @throws Exception
     */
    @Override
    public void subscribe(URL url) throws Exception {
        // 服务监听路径
        final String path = getProviderPath(url.getServiceName(), url.getVersion());
        // 添加服务提供方资源定位器
        Cache.SUBSCRIBE_SERVICE_LIST.add(url);
        // 开启服务监听
        this.watchNodeDataChange(path);
    }

    /**
     * 取消服务订阅
     *
     * @param url
     */
    @Override
    public void unSubscribe(URL url) {
        // 服务路径
        final String path = getProviderPath(url.getServiceName(), url.getVersion());
        // 删除服务提供方网络资源定位器
        Cache.SUBSCRIBE_SERVICE_LIST.remove(url);
        // 关闭服务监听
        try {
            PathChildrenCache cache = PATH_CHILDREN_CACHE_MAP.get(path);
            cache.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 服务监听，监听服务节点数据的更新
     * @param path 服务路径
     * @throws Exception
     */
    private void watchNodeDataChange(String path) throws Exception {
        // 创建监听缓存
        PathChildrenCache cache = new PathChildrenCache(client, path, true);

        // 添加PathChildrenCacheListener监听器
        cache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                // 获取监听事件类型
                final PathChildrenCacheEvent.Type type = event.getType();
                System.out.println("PathChildrenCache event: " + type);
                // 服务事件
                RpcEventData eventData = null;
                // 销毁事件类型
                if (type.equals(CHILD_REMOVED)) {
                    // 获取服务节点的路径
                    String path = event.getData().getPath();
                    // 创建服务节点的服务提供方资源定位器
                    final URL url = parsePath(path);
                    eventData = new DestroyRpcEventData(url);
                } else if ((type.equals(CHILD_UPDATED)) || type.equals(CHILD_ADDED)) {
                    // 获取服务节点的路径
                    String path = event.getData().getPath();
                    byte[] bytes = client.getData().forPath(path);
                    Object o = JSON.parseObject(bytes, URL.class);
                    eventData = type.equals(CHILD_UPDATED) ? new UpdateRpcEventData(o) : new AddRpcEventData(o);
                }
                // 发送服务事件
                RpcListerLoader.sendEvent(eventData);
            }
        });

        // 启动PathChildrenCache，监听缓存初始完成后触发监听事件
        cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);

        // 添加监听
        PATH_CHILDREN_CACHE_MAP.put(path,cache);
    }


    // 获取服务节点路径
    private String getProviderDataPath(URL url) {
        // ROOT_PATH + PROVIDER + Service + $ version + Host + Port : 节点
        return ROOT_PATH + PROVIDER + "/" + url.getServiceName() + "/" + url.getVersion() + "/" + url.getIP() + ":" + url.getPort();
    }

    // 获取服务监听路径
    private String getProviderPath(URL url) {
        return ROOT_PATH + PROVIDER + "/" + url.getServiceName() + "/" + url.getVersion();
    }

    // 获取服务监听路径
    private String getProviderPath(String serviceName, String version) {
        return ROOT_PATH + PROVIDER + "/" + serviceName + "/" + version;
    }

    // 路径解析
    private URL parsePath(String path) {
        final String[] split = path.split("/");
        String serviceName = split[3];
        String version = split[4];
        final String[] split1 = split[5].split(":");
        String Ip = split1[0];
        String port = split[1];
        final URL url = new URL();
        url.setIP(Ip);
        url.setPort(Integer.getInteger(port));
        url.setVersion(version);
        url.setServiceName(serviceName);
        return url;
    }

    // 删除节点
    private boolean deleteNode(String path) {
        try {
            client.delete().forPath(path);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }


    // 判断存在节点
    private boolean existNode(String path) {
        try {
            Stat stat = client.checkExists().forPath(path);
            return stat != null;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }
}
